package com.atguigu.gmall2020.realtime.spark.app

import java.lang
import java.text.SimpleDateFormat
import java.util.Date

import com.alibaba.fastjson.{JSON, JSONObject}
import com.atguigu.gmall2020.realtime.spark.util.RedisUtil
import com.atguigu.gmall2020.realtime.spark.util.KafkaUtil
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{HasOffsetRanges, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import redis.clients.jedis.Jedis

import scala.collection.mutable.ListBuffer

object DauTest {

  //1、要从日志中择出访问信息，（如果只是app的话，可以去应用的启动日志 如果还有页面或者链接 取访问的第一个页面的访问日志 ）根据last_page_id 判断 如果为空 则表示该用户本次访问的第一个页面

  //2、 对多次访问的首页日志，进行去重，redis
  //3、写入到es 中
  def main(args: Array[String]): Unit = {

    //1.创建spark配置文件对象,指定核心数,这里指定的core需要考虑和Kafka中的分区数适配,尽可能的做到并行处理.
    val sparkConf = new SparkConf().setMaster("local[4]").setAppName("DauApp")
    //2.初始化SparkStreamingContext,指定配置对象并指定多久传输周期
    val ssc = new StreamingContext(sparkConf, Seconds(5))
    //3.通过我们自己定义的工具类,通过Kafka数据源创建一个离散化流,处理我们存入到Kafka的主题中的用户行为日志的数据.
    val topicName = "ODS_BASE_LOG"
    val groupId = "dau_group"
    val KafkaDS: InputDStream[ConsumerRecord[String, String]] = KafkaUtil.getKafkaStream(topicName, ssc, groupId)

    //1、要从日志中择出访问信息，（如果只是app的话，可以去应用的启动日志 如果还有页面或者链接 取访问的第一个页面的访问日志 ）
    // 根据last_page_id 判断 如果为空 则表示该用户本次访问的第一个页面   //筛选
    //1.1 为了筛选操作方便 首先可以把字符串转为方便处理json对象
    // json string==>1、专用的结构化对象： bean或case class 2  通用的结构化对象 map 或者jsonObject
    val jsonDStream: DStream[JSONObject] = KafkaDS.map { record =>
      val jsonString: String = record.value()
      val jSONObject: JSONObject = JSON.parseObject(jsonString)
      //补充对日期加工  ts -> hr ts->yyyy-MM-dd
      val ts: lang.Long = jSONObject.getLong("ts")

      val dateFormat = new SimpleDateFormat("yyyy-MM-dd HH")
      val dateHrString: String = dateFormat.format(new Date(ts))
      val dataHrArr: Array[String] = dateHrString.split(" ")
      val dt: String = dataHrArr(0)
      val hr: String = dataHrArr(1)
      jSONObject.put("dt", dt)
      jSONObject.put("hr", hr)
      jSONObject
    }
    //1.2 根据last_page_id 判断 如果为空 则表示该用户本次访问的第一个页面
    // 筛选
    val firstDStream: DStream[JSONObject] = jsonDStream.filter { jSONObject =>
      var ifFirst = false
      val pageObj: JSONObject = jSONObject.getJSONObject("page")
      if (pageObj != null) {
        val lastPageId: String = pageObj.getString("last_page_id")
        if (lastPageId != null && lastPageId.size > 0) {
          ifFirst = false
        } else {
          ifFirst = true //只有包含访问的日志 且该访问没有上一页  则视为首页保留
        }
      }
      ifFirst
    }

    val dauDstream: DStream[JSONObject] = firstDStream.mapPartitions { jsonObjItr => //每分区 每批次的数据
      val jedis: Jedis = RedisUtil.getJedisClient //分区共用
      val targerList: ListBuffer[JSONObject] = new ListBuffer[JSONObject]
      val originList: List[JSONObject] = jsonObjItr.toList
      println("该分区共：" + originList.size + "条")
      for (jsonObj <- originList) { //每条数据
        val commonJSONObj: JSONObject = jsonObj.getJSONObject("common") //executor
        val mid: String = commonJSONObj.getString("mid")
        val dt: String = jsonObj.getString("dt")
        val key = "dau:" + dt

        val isNonExistsFlag: lang.Long = jedis.sadd(key, mid) // sadd  既可以进行判断 有可以写入  原子性命令
        if (isNonExistsFlag == 1L) {
          targerList.append(jsonObj)
        }
      }
      jedis.close()
      println("该分区筛选后 共：" + targerList.size + "条")
      targerList.toIterator
    }
    // dr   // 启动时执行一次
    //    dauDstream.transform{rdd=>
    //      // dr ? driver     //  周期性
    //      rdd.map{
    //        null  //ex
    //      }
    //      null
    //    }
    dauDstream.print(1000)
    ssc.start()
    ssc.awaitTermination()

  }

}
